「なぜ料金が高くなった？」AWSコスト分析をAmazon Bedrockに丸投げしてSlack通知する

AWSの利用料を生成AIに分析させてコスト管理してみた
できること
構成
コード
設定
注意点
完成！
おわりに

AWSの利用料を生成AIに分析させてコスト管理してみた

こんにちは！

コンソールにログインして、AWS Cost Explorer（以下、Cost Explorer）の画面に行って･･･
AWSのコスト管理は正直めんどくさいですよね！
利用料をSlackなどに投稿することはできますが、毎日数字やグラフとにらめっこするのは大変です。

今回は、AWSの生成AIサービスである「Amazon Bedrock（以下、Bedrock）」を利用して、
「生成AIにコストが増加している原因や傾向を分析させて、その要約をSlackに投稿する」仕組みを作りました。

できること

この記事で紹介する仕組みを実装すると、以下のことができるようになります。

AWSのコストが増加しているサービスを生成AIに分析させ、特定する
コストが増加したサービスごとにどれだけ増えているか差額を出力する
1日のコストの閾値を設定しておき、閾値を超えた日を特定する
結果をSlackの特定のチャンネルに投稿する

コストの分析期間は7日前～2日前です。
1日前＆当日はすべての利用料がCost Explorerに反映されていません。
Slack投稿は毎日行われます。

構成

構成は以下の通りです。

以下の処理をAmazon EventBridge Schedulerで毎日定時実行
- Cost Explorer APIで7日前～2日前のAWS利用料を取得
- Bedrock APIでコスト分析

シンプルですね！
Bedrockのモデルは「Amazon Nova Pro」を利用します。

コード

以下にLambdaのコードを記載します。

import os
import json
import logging
import datetime
from zoneinfo import ZoneInfo
import urllib.request
import urllib.parse

import boto3

# --- 定数定義 ---

# 日本標準時 （UTC+9）のタイムゾーン情報
JST                       = datetime.timezone(datetime.timedelta(hours=+9), 'JST')

# 実行時の現在時刻（JST）
DATETIME_NOW              = datetime.datetime.now(JST)

# AWSコスト利用料の閾値
THRESHOLD_BOLD_TOTAL_DIFF = 50.0

# Slack投稿文のヘッダー部分
SLACK_POST_MESSAGES_HEADER_LIST = [str(DATETIME_NOW).replace("-", "/").split(".")[0] + " 時点でのAWS利用料の分析です。", ""]

# Slack投稿に必要な情報、後述
SLACK_BOT_TOKEN = "your_slack_bot_token"
SLACK_WEBHOOK_URL = "your_slack_bot_webhook_url"
SLACK_CHANNEL_ID = "your_slack_channel_id"

# 使用するBedrockのモデルID
BEDROCK_MODEL_ID = os.environ.get('BEDROCK_MODEL_ID', 'apac.amazon.nova-pro-v1:0')

# ログ出力の設定
logger = logging.getLogger()
logger.setLevel(logging.INFO)

def index_of(target_list, target):
"""指定された要素がリスト内のどこに位置するかを取得する関数"""
    if (target in target_list):
        return target_list.index(target)
    else:
        return -1

def generate_post_message():
"""Slackに投稿するメッセージを生成する関数"""

    post_messages_list = list(SLACK_POST_MESSAGES_HEADER_LIST)

    ai_analysis_result_text = analyze_from_ai(THRESHOLD_BOLD_TOTAL_DIFF)
    post_messages_list.insert(index_of(post_messages_list, ""), "AIによる分析：" + ai_analysis_result_text)

    return post_messages_list

def post_messages_to_slack(message):
"""Slackにメッセージを投稿する関数"""

    # アタッチメントの色の変更
    attachment_color = "good"
    
    # メッセージの送信
    send_data = {
        "token"       : SLACK_BOT_TOKEN,
        "channel"     : SLACK_CHANNEL_ID,
        "attachments" : [
            {
                "color" : attachment_color,
                "text"  : message
            }
        ]
    }
    send_text = "payload=" + json.dumps(send_data)
    
    request = urllib.request.Request(
        SLACK_WEBHOOK_URL, 
        data=send_text.encode("utf-8"), 
        method="POST"
    )
    
    with urllib.request.urlopen(request) as response:
        response_body = response.read().decode("utf-8")

def analyze_from_ai(cost_threshold):
"""日付を計算して、該当範囲の分析を依頼する関数"""
    
    # JST正午時点での「昨日(JST)」と「7日前(JST)」の日付を計算
    jst_today = DATETIME_NOW
    jst_yesterday_dt = jst_today - datetime.timedelta(days=1)
    jst_7days_ago_dt = jst_today - datetime.timedelta(days=7)
    
    start_date = jst_7days_ago_dt.strftime('%Y-%m-%d')
    # Cost ExplorerのEndは含まれないため、当日の日付を指定する
    end_date =jst_yesterday_dt.strftime('%Y-%m-%d')
    
    # Cost Explorerで7日前～2日前のデータを取得
    try:
        # データ1: 日ごとの合計料金
        daily_total_costs = get_daily_total_costs(start_date, end_date)
        
        # データ2: 日ごとのサービスごとの料金
        daily_service_costs = get_daily_service_costs(start_date, end_date)

    except Exception as e:
        return "Error: 料金データの取得に失敗しました。\n{e}"

    # Bedrockに分析を依頼
    ai_analysis_result = analyze_weekly_costs_by_ai(
        daily_total_costs, 
        daily_service_costs, 
        cost_threshold
    )

    return ai_analysis_result

def get_daily_total_costs(start_date, end_date):  
"""指定した期間の日ごとの合計金額を計算する関数"""

    client = boto3.client("ce", region_name='ap-northeast-1')

    response = client.get_cost_and_usage(
        TimePeriod={'Start': start_date, 'End': end_date},
        Granularity='DAILY',
        # クレジットが適用されている場合に料金が引かれて算出されるため、除外しておく
        Filter={
            "Not" : { # 除外設定
                "Dimensions": {
                    "Key"    : "RECORD_TYPE",
                    "Values" : [
                        "Credit" # クレジット
                    ]
                }
            }
        },
        Metrics=['UnblendedCost']
        # GroupBy を指定しない = 合計を取得
    )
    
    costs = {}
    for day_data in response['ResultsByTime']:
        date = day_data['TimePeriod']['Start']
        cost = float(day_data['Total']['UnblendedCost']['Amount'])
        costs[date] = cost
    return costs


def get_daily_service_costs(start_date, end_date):
"""指定した期間の日ごとのサービごとの料金を計算する関数"""

    client = boto3.client("ce", region_name='ap-northeast-1')

    response = client.get_cost_and_usage(
        TimePeriod={'Start': start_date, 'End': end_date},
        Granularity='DAILY',
        # クレジットが適用されている場合に料金が引かれて算出されるため、除外しておく
        Filter={
            "Not" : { # 除外設定
                "Dimensions": {
                    "Key"    : "RECORD_TYPE",
                    "Values" : [
                        "Credit" # クレジット
                    ]
                }
            }
        },
        Metrics=['UnblendedCost'],
        GroupBy=[
            {'Type': 'DIMENSION', 'Key': 'SERVICE'}
        ]
    )
    
    # { '日付': {'サービス名': 金額, ...}, ... } の形式に整形
    costs_by_date = {}
    for day_data in response['ResultsByTime']:
        date = day_data['TimePeriod']['Start']
        service_costs = {}
        for group in day_data['Groups']:
            service_name = group['Keys'][0]
            cost = float(group['Metrics']['UnblendedCost']['Amount'])
            if cost > 0:
                service_costs[service_name] = cost
        costs_by_date[date] = service_costs
    return costs_by_date

def analyze_weekly_costs_by_ai(daily_total_costs, daily_service_costs, threshold):
"""AIを呼び出して料金を分析する関数"""

    # AIへの入力（データ）を整形
    prompt_data = "【7日前から2日前の日ごと合計料金】\n"
    for date, cost in sorted(daily_total_costs.items()):
        prompt_data += f"- {date}: ${cost:,.2f}\n"
    
    prompt_data += "\n【7日前から2日前のサービスごと料金詳細】\n"
    for date, services in sorted(daily_service_costs.items()):
        prompt_data += f"--- {date} ---\n"
        if not services:
            prompt_data += " (データなし)\n"
            continue
        # 金額の多い順にソートして表示
        for service, cost in sorted(services.items(), key=lambda item: item[1], reverse=True):
            if cost > 0.01: # わずかな金額は省略
                prompt_data += f"  - {service}: ${cost:,.2f}\n"


    # プロンプトの構築
    prompt = f"""Human:
あなたはAWSコスト分析の専門家です。
以下の料金データに基づき、2つのタスクを実行してください。

【タスク】
1.【7日前から2日前の日ごと合計料金】を参照し、閾値 (${threshold}) を超えている日がないか確認してください。
2.【7日前から2日前のサービスごと料金詳細】を参照し、この1週間で特にコストが1$以上増加しているAWSサービスを特定し、その傾向（例: どのサービスが何日から何日にかけて増加したか）を簡潔に分析してください。

【データ】
{prompt_data}

【分析結果】
分析結果を箇条書きでまとめてください。
閾値については、閾値自体の記載および、閾値を超えた日のみ日付けと金額を記載してください。
サービスごと料金詳細については、改善傾向や推奨アクションは必要なく、増加しているサービスの傾向だけを1~2行で具体的な数字を含めて簡潔にまとめてください。
増加していないサービスは記載しないでください。
最後にまとめを記載する必要はありません。

Assistant:
"""

    client = boto3.client('bedrock-runtime')

    # Bedrock APIの呼び出し
    body = json.dumps({
        "messages": [
            {
                "role": "user",
                "content": [
                    {
                        "text":prompt
                    }
                ]
            }
        ],
        "inferenceConfig": {
            "maxTokens": 2000
        }
    })
    
    response = client.invoke_model(
        body=body,
        modelId=BEDROCK_MODEL_ID,
        contentType='application/json',
        accept='*/*'
    )
    
    response_body = json.loads(response['body'].read())
    analysis_text = response_body['output']['message']['content'][0]['text']
    
    return analysis_text

def lambda_handler(event, context):
"""Lambdaのメイン関数"""

    try:
         # Slackへ投稿するメッセージの生成
        post_messages_list = generate_post_message()
        logger.info("Slackへ投稿するメッセージの生成完了 post_messages_list = " + str(post_messages_list))
        
        # Slackへ投稿
        post_messages_to_slack("\n".join(post_messages_list))
        logger.info("Slackへのメッセージ投稿完了")
    except Exception as e:
        logger.error(e)
        raise e

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

import os

import json

import logging

import datetime

from zoneinfo import ZoneInfo

import urllib.request

import urllib.parse

import boto3

# --- 定数定義 ---

# 日本標準時（UTC+9）のタイムゾーン情報

JST = datetime.timezone(datetime.timedelta(hours=+9), 'JST')

# 実行時の現在時刻（JST）

DATETIME_NOW = datetime.datetime.now(JST)

# AWSコスト利用料の閾値

THRESHOLD_BOLD_TOTAL_DIFF = 50.0

# Slack投稿文のヘッダー部分

SLACK_POST_MESSAGES_HEADER_LIST = [str(DATETIME_NOW).replace("-", "/").split(".")[0] + " 時点でのAWS利用料の分析です。", ""]

# Slack投稿に必要な情報、後述

SLACK_BOT_TOKEN = "your_slack_bot_token"

SLACK_WEBHOOK_URL = "your_slack_bot_webhook_url"

SLACK_CHANNEL_ID = "your_slack_channel_id"

# 使用するBedrockのモデルID

BEDROCK_MODEL_ID = os.environ.get('BEDROCK_MODEL_ID', 'apac.amazon.nova-pro-v1:0')

# ログ出力の設定

logger = logging.getLogger()

logger.setLevel(logging.INFO)

def index_of(target_list, target):

"""指定された要素がリスト内のどこに位置するかを取得する関数"""

if (target in target_list):

return target_list.index(target)

else:

return -1

def generate_post_message():

"""Slackに投稿するメッセージを生成する関数"""

post_messages_list = list(SLACK_POST_MESSAGES_HEADER_LIST)

ai_analysis_result_text = analyze_from_ai(THRESHOLD_BOLD_TOTAL_DIFF)

post_messages_list.insert(index_of(post_messages_list, ""), "AIによる分析：" + ai_analysis_result_text)

return post_messages_list

def post_messages_to_slack(message):

"""Slackにメッセージを投稿する関数"""

# アタッチメントの色の変更

attachment_color = "good"

# メッセージの送信

send_data = {

"token" : SLACK_BOT_TOKEN,

"channel" : SLACK_CHANNEL_ID,

"attachments" : [

{

"color" : attachment_color,

"text" : message

}

]

}

send_text = "payload=" + json.dumps(send_data)

request = urllib.request.Request(

SLACK_WEBHOOK_URL,

data=send_text.encode("utf-8"),

method="POST"

)

with urllib.request.urlopen(request) as response:

response_body = response.read().decode("utf-8")

def analyze_from_ai(cost_threshold):

"""日付を計算して、該当範囲の分析を依頼する関数"""

# JST正午時点での「昨日(JST)」と「7日前(JST)」の日付を計算

jst_today = DATETIME_NOW

jst_yesterday_dt = jst_today - datetime.timedelta(days=1)

jst_7days_ago_dt = jst_today - datetime.timedelta(days=7)

start_date = jst_7days_ago_dt.strftime('%Y-%m-%d')

# Cost ExplorerのEndは含まれないため、当日の日付を指定する

end_date =jst_yesterday_dt.strftime('%Y-%m-%d')

# Cost Explorerで7日前～2日前のデータを取得

try:

# データ1: 日ごとの合計料金

daily_total_costs = get_daily_total_costs(start_date, end_date)

# データ2: 日ごとのサービスごとの料金

daily_service_costs = get_daily_service_costs(start_date, end_date)

except Exception as e:

return "Error: 料金データの取得に失敗しました。\n{e}"

# Bedrockに分析を依頼

ai_analysis_result = analyze_weekly_costs_by_ai(

daily_total_costs,

daily_service_costs,

cost_threshold

)

return ai_analysis_result

def get_daily_total_costs(start_date, end_date):

"""指定した期間の日ごとの合計金額を計算する関数"""

client = boto3.client("ce", region_name='ap-northeast-1')

response = client.get_cost_and_usage(

TimePeriod={'Start': start_date, 'End': end_date},

Granularity='DAILY',

# クレジットが適用されている場合に料金が引かれて算出されるため、除外しておく

Filter={

"Not" : { # 除外設定

"Dimensions": {

"Key" : "RECORD_TYPE",

"Values" : [

"Credit" # クレジット

]

}

Metrics=['UnblendedCost']

# GroupBy を指定しない = 合計を取得

)

costs = {}

for day_data in response['ResultsByTime']:

date = day_data['TimePeriod']['Start']

cost = float(day_data['Total']['UnblendedCost']['Amount'])

costs[date] = cost

return costs

def get_daily_service_costs(start_date, end_date):

"""指定した期間の日ごとのサービごとの料金を計算する関数"""

client = boto3.client("ce", region_name='ap-northeast-1')

response = client.get_cost_and_usage(

TimePeriod={'Start': start_date, 'End': end_date},

Granularity='DAILY',

# クレジットが適用されている場合に料金が引かれて算出されるため、除外しておく

Filter={

"Not" : { # 除外設定

"Dimensions": {

"Key" : "RECORD_TYPE",

"Values" : [

"Credit" # クレジット

]

}

Metrics=['UnblendedCost'],

GroupBy=[

{'Type': 'DIMENSION', 'Key': 'SERVICE'}

]

)

# { '日付': {'サービス名': 金額, ...}, ... } の形式に整形

costs_by_date = {}

for day_data in response['ResultsByTime']:

date = day_data['TimePeriod']['Start']

service_costs = {}

for group in day_data['Groups']:

service_name = group['Keys'][0]

cost = float(group['Metrics']['UnblendedCost']['Amount'])

if cost > 0:

service_costs[service_name] = cost

costs_by_date[date] = service_costs

return costs_by_date

def analyze_weekly_costs_by_ai(daily_total_costs, daily_service_costs, threshold):

"""AIを呼び出して料金を分析する関数"""

# AIへの入力（データ）を整形

prompt_data = "【7日前から2日前の日ごと合計料金】\n"

for date, cost in sorted(daily_total_costs.items()):

prompt_data += f"- {date}: ${cost:,.2f}\n"

prompt_data += "\n【7日前から2日前のサービスごと料金詳細】\n"

for date, services in sorted(daily_service_costs.items()):

prompt_data += f"--- {date} ---\n"

if not services:

prompt_data += " (データなし)\n"

continue

# 金額の多い順にソートして表示

for service, cost in sorted(services.items(), key=lambda item: item[1], reverse=True):

if cost > 0.01: # わずかな金額は省略

prompt_data += f" - {service}: ${cost:,.2f}\n"

# プロンプトの構築

prompt = f"""Human:

あなたはAWSコスト分析の専門家です。

以下の料金データに基づき、2つのタスクを実行してください。

【タスク】

1.【7日前から2日前の日ごと合計料金】を参照し、閾値 (${threshold}) を超えている日がないか確認してください。

2.【7日前から2日前のサービスごと料金詳細】を参照し、この1週間で特にコストが1$以上増加しているAWSサービスを特定し、その傾向（例: どのサービスが何日から何日にかけて増加したか）を簡潔に分析してください。

【データ】

{prompt_data}

【分析結果】

分析結果を箇条書きでまとめてください。

閾値については、閾値自体の記載および、閾値を超えた日のみ日付けと金額を記載してください。

サービスごと料金詳細については、改善傾向や推奨アクションは必要なく、増加しているサービスの傾向だけを1~2行で具体的な数字を含めて簡潔にまとめてください。

増加していないサービスは記載しないでください。

最後にまとめを記載する必要はありません。

Assistant:

"""

client = boto3.client('bedrock-runtime')

# Bedrock APIの呼び出し

body = json.dumps({

"messages": [

{

"role": "user",

"content": [

{

"text":prompt

}

]

}

"inferenceConfig": {

"maxTokens": 2000

}

})

response = client.invoke_model(

body=body,

modelId=BEDROCK_MODEL_ID,

contentType='application/json',

accept='*/*'

)

response_body = json.loads(response['body'].read())

analysis_text = response_body['output']['message']['content'][0]['text']

return analysis_text

def lambda_handler(event, context):

"""Lambdaのメイン関数"""

try:

# Slackへ投稿するメッセージの生成

post_messages_list = generate_post_message()

logger.info("Slackへ投稿するメッセージの生成完了 post_messages_list = " + str(post_messages_list))

# Slackへ投稿

post_messages_to_slack("\n".join(post_messages_list))

logger.info("Slackへのメッセージ投稿完了")

except Exception as e:

logger.error(e)

raise e

自分で設定する必要のある変数は以下の通りです。

THRESHOLD_BOLD_TOTAL_DIFF：AWSコスト利用料の閾値。
- この値を超えている日があるかどうかAIが分析してくれます。
  　自身のAWS利用想定や規模に応じて設定してください。
SLACK_BOT_TOKEN：投稿に利用するSlack Botのトークン。
SLACK_WEBHOOK_URL：投稿に利用するSlack BotのWebhook URL。
- SLACK_BOT_TOKENと併せて、確認方法はこちらを参照してください。
SLACK_CHANNEL_ID：投稿するSlack ChannelのID。
- 確認方法はこちらを参照してください。

また、上記の「SLACK_」で始まる3つの変数について、
今回は動作をシンプルにイメージできるように直接コード上に記載していますが、実際はハードコーディングをしないようにしてください。
Lambdaの環境変数に設定したり、Parameter Storeに保存したりして保守性を上げましょう。

設定

Lambda関数の設定は以下の通りです。

ランタイム：Python 3.14
アーキテクチャ：x86_64
メモリ：128MB
タイムアウト：1分0秒
- 利用しているサービスの多さによって、Bedrockでの分析に数十秒かかります。
EventBridgeのスケジュール式：cron(0 X ? * * *)
- Xの部分に0~15の数字を記載します。
- 式は世界標準時を表しているので、+9して日本時間を計算できます。
  　（例：cron(0 3 ? * * *)なら日本時間の正午）
実行ロール：lambda_basic_executionの他に、以下のポリシーが必要です。

Cost Explorerの読み取りポリシー

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": [
                "ce:DescribeCostCategoryDefinition",
                "ce:GetRightsizingRecommendation",
                "ce:GetCostAndUsage",
                "ce:GetSavingsPlansUtilization",
                "ce:GetReservationPurchaseRecommendation",
                "ce:ListCostCategoryDefinitions",
                "ce:GetCostForecast",
                "ce:GetReservationUtilization",
                "ce:GetSavingsPlansPurchaseRecommendation",
                "ce:GetDimensionValues",
                "ce:GetSavingsPlansUtilizationDetails",
                "ce:GetCostAndUsageWithResources",
                "ce:GetReservationCoverage",
                "ce:GetSavingsPlansCoverage",
                "ce:GetTags",
                "ce:GetUsageForecast",
                "cur:DescribeReportDefinitions"
            ],
            "Resource": "*"
        }
    ]
}

{

"Version": "2012-10-17",

"Statement": [

{

"Sid": "VisualEditor0",

"Effect": "Allow",

"Action": [

"ce:DescribeCostCategoryDefinition",

"ce:GetRightsizingRecommendation",

"ce:GetCostAndUsage",

"ce:GetSavingsPlansUtilization",

"ce:GetReservationPurchaseRecommendation",

"ce:ListCostCategoryDefinitions",

"ce:GetCostForecast",

"ce:GetReservationUtilization",

"ce:GetSavingsPlansPurchaseRecommendation",

"ce:GetDimensionValues",

"ce:GetSavingsPlansUtilizationDetails",

"ce:GetCostAndUsageWithResources",

"ce:GetReservationCoverage",

"ce:GetSavingsPlansCoverage",

"ce:GetTags",

"ce:GetUsageForecast",

"cur:DescribeReportDefinitions"

"Resource": "*"

}

]

}

Bedrockの呼び出しポリシー

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": "bedrock:InvokeModel",
            "Resource": "*"
        }
    ]
}

{

"Version": "2012-10-17",

"Statement": [

{

"Sid": "VisualEditor0",

"Effect": "Allow",

"Action": "bedrock:InvokeModel",

"Resource": "*"

}

]

}

注意点

「できること」でも軽く触れましたが、
この仕組みでは、昨日・今日の直近のコストは分析していません。
Cost Explorerに利用料が完全に反映されておらず、一部しか表示されないからです。
直近の異常利用などを検出したいときは、Anomaly Detectionなど別の仕組みを利用してください。

また、分析を完全に信用しきることはできません。生成AIはハルシネーションを起こす可能性があります。
この仕組みだけで利用料の監視を終わらせずに、違和感があるときは必ずCost Explorerや請求を確認してください。